package com.gradle.producer.kafka.config;

import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;

import java.util.Map;

/**
 * 自定义分区
 *
 * @author ldanniel
 */
public class CustomizePartitioner implements Partitioner {
    /**
     * //自定义分区规则，默认全部发送到0号分区
     *
     * @param topic      主题
     * @param key        序列化之前的key
     * @param keyBytes   序列化之后的key
     * @param value      序列化之前的value
     * @param valueBytes 序列化之后的value
     * @param cluster    cluster则是集群信息（比如当前 Kafka 集群共有多少主题、多少 Broker 等）
     * @return partition
     */
    @Override
    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
        if (value.toString().contains("tx")) {
            return 1;
        }
        return 0;
    }

    @Override
    public void close() {

    }

    @Override
    public void configure(Map<String, ?> configs) {

    }
}
